package com.bw.app.dwd1;

import com.bw.utils.MyKafkaUtil;
import com.bw.utils.MysqlUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.ZoneId;


/**
 * @projectName: Flinksxe
 * @fileName: DwdTradeOrderPreProcess
 * @packageName: com.bw.app.dwd
 * @date: 2025/9/15 19:08
 * @author: ymw
 */
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
//        Configuration configuration = tenv.getConfig().getConfiguration();
//        configuration.setString("table.exec.state.ttl", "5s");
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1L),Time.minutes(1L)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
//        System.setProperty("HADOOP_USER_NAME","bw");
//        tenv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));
        tenv.executeSql("create table topic_db(\n" +
                "    `database` string,\n" +
                "    `table` string,\n" +
                "    `type` string,\n" +
                "    `data` map<string,string>,\n" +
                "    `old` map<string,string>,\n" +
                "    `ts` string,\n" +
                "    pt as proctime()\n" +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","dwd_trade_order_pre_process"));
        Table orderDetailTable = tenv.sqlQuery("select\n" +
                "    `data`['id'] as id,\n" +
                "    `data`['order_id'] as order_id,\n" +
                "    `data`['sku_id'] as sku_id,\n" +
                "    `data`['sku_name'] as sku_name,\n" +
                "    `data`['order_price'] as order_price,\n" +
                "    `data`['sku_num'] as sku_num,\n" +
                "    `data`['create_time'] as create_time,\n" +
                "    `data`['source_type'] as source_type,\n" +
                "    `data`['source_id'] as source_id,\n" +
                "    `data`['split_total_amount'] as split_total_amount,\n" +
                "    `data`['split_activity_amount'] as split_activity_amount,\n" +
                "    `data`['split_coupon_amount'] as split_coupon_amount,\n" +
                "     pt\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                " and  `table` = 'order_detail'\n" +
                " and `type` = 'insert' ");
        tenv.createTemporaryView("order_detail_table",orderDetailTable);
//        tenv.sqlQuery("select * from order_detail_table").execute().print();
        tenv.toAppendStream(orderDetailTable, Row.class).print("order_detail====>");

        Table orderInfo = tenv.sqlQuery("select\n" +
                "    data['id'] as id,\n" +
                "    data['consignee'] as consignee,\n" +
                "    data['consignee_tel'] as consignee_tel,\n" +
                "    data['total_amount'] as total_amount,\n" +
                "    data['order_status'] as order_status,\n" +
                "    data['user_id'] as user_id,\n" +
                "    data['payment_way'] as payment_way,\n" +
                "    data['delivery_address'] as delivery_address,\n" +
                "    data['order_comment'] as order_comment,\n" +
                "    data['out_trade_no'] as out_trade_no,\n" +
                "    data['trade_body'] as trade_body,\n" +
                "    data['create_time'] as create_time,\n" +
                "    data['operate_time'] as operate_time,\n" +
                "    data['expire_time'] as expire_time,\n" +
                "    data['process_status'] as process_status,\n" +
                "    data['tracking_no'] as tracking_no,\n" +
                "    data['parent_order_id'] as parent_order_id,\n" +
                "    data['province_id'] as province_id,\n" +
                "    data['activity_reduce_amount'] as activity_reduce_amount,\n" +
                "    data['coupon_reduce_amount'] as coupon_reduce_amount,\n" +
                "    data['original_total_amount'] as original_total_amount,\n" +
                "    data['feight_fee'] as feight_fee,\n" +
                "    data['feight_fee_reduce'] as feight_fee_reduce,\n" +
                "    data['refundable_time'] as refundable_time,\n" +
                "    `type`,\n" +
                "    `old`\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'order_info'\n" +
                "and (`type` = 'insert' or `type` = 'update')");
        tenv.createTemporaryView("order_info_table",orderInfo);
        tenv.toAppendStream(orderInfo, Row.class).print("order_info_table=====>");
        Table orderDetailActivity = tenv.sqlQuery("select\n" +
                "    data['id'] as id,\n" +
                "    data['order_id'] as order_id,\n" +
                "    data['order_detail_id'] as order_detail_id,\n" +
                "    data['activity_id'] as activity_id,\n" +
                "    data['activity_rule_id'] as activity_rule_id,\n" +
                "    data['sku_id'] as sku_id,\n" +
                "    data['create_time'] as create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and  `table` = 'order_detail_activity'\n" +
                "and `type` = 'insert'");
        tenv.createTemporaryView("order_detail_activity_table",orderDetailActivity);
        tenv.toAppendStream(orderDetailActivity, Row.class).print("order_detail_activity====》");
        Table orderDetailCoupon = tenv.sqlQuery("select\n" +
                "  data['id'] as id,\n" +
                "  data['order_id'] as order_id,\n" +
                "  data['order_detail_id'] as order_detail_id,\n" +
                "  data['coupon_id'] as coupon_id,\n" +
                "  data['coupon_use_id'] as coupon_use_id,\n" +
                "  data['sku_id'] as sku_id,\n" +
                "  data['create_time'] as create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and  `table` = 'order_detail_coupon'\n" +
                "and `type` = 'insert'");
        tenv.createTemporaryView("order_detail_coupon_table",orderDetailCoupon);
        tenv.toAppendStream(orderDetailCoupon, Row.class).print("order_detail_coupon====>");

        tenv.executeSql("create table base_dic(\n" +
                "    dic_code string,\n" +
                "    dic_name string,\n" +
                "    parent_code string,\n" +
                "    create_time string,\n" +
                "    operate_time string,\n" +
                "    primary key(`dic_code`) not enforced\n" +
                ")" + MysqlUtil.mysqlLookUpTableDDL("base_dic"));
        Table resultTable = tenv.sqlQuery("select\n" +
                "    od.id,\n" +
                "    od.order_id,\n" +
                "    od.sku_id,\n" +
                "    od.sku_name,\n" +
                "    od.order_price,\n" +
                "    od.sku_num,\n" +
                "    od.create_time,\n" +
                "    od.source_type as source_type_id,\n" +
                "    b.dic_name as source_type_name,\n" +
                "    od.source_id,\n" +
                "    od.split_total_amount,\n" +
                "    od.split_activity_amount,\n" +
                "    od.split_coupon_amount,\n" +
                "    oi.consignee,\n" +
                "    oi.consignee_tel,\n" +
                "    oi.total_amount,\n" +
                "    oi.order_status,\n" +
                "    oi.user_id,\n" +
                "    oi.payment_way,\n" +
                "    oi.delivery_address,\n" +
                "    oi.order_comment,\n" +
                "    oi.out_trade_no,\n" +
                "    oi.trade_body,\n" +
                "    oi.operate_time,\n" +
                "    oi.expire_time,\n" +
                "    oi.process_status,\n" +
                "    oi.tracking_no,\n" +
                "    oi.parent_order_id,\n" +
                "    oi.province_id,\n" +
                "    oi.activity_reduce_amount,\n" +
                "    oi.coupon_reduce_amount,\n" +
                "    oi.original_total_amount,\n" +
                "    oi.feight_fee,\n" +
                "    oi.feight_fee_reduce,\n" +
                "    oi.refundable_time,\n" +
                "    oda.id as order_detail_activity_id,\n" +
                "    oda.activity_id,\n" +
                "    oda.activity_rule_id,\n" +
                "    odc.id as order_detail_coupon_id,\n" +
                "    odc.coupon_id,\n" +
                "    odc.coupon_use_id,\n" +
                "    oi.`type` ,\n" +
                "    oi.`old`,\n" +
                "    current_row_timestamp() as row_op_ts\n" +
                "from order_detail_table od\n" +
                "join order_info_table oi\n" +
                "on od.order_id = oi.id\n" +
                "left join order_detail_activity_table oda\n" +
                "on od.id = oda.order_detail_id\n" +
                "left join order_detail_coupon_table odc\n" +
                "on od.id = odc.order_detail_id\n" +
                "JOIN base_dic FOR SYSTEM_TIME AS OF od.pt AS b\n" +
                "ON od.source_type = b.dic_code");
        tenv.createTemporaryView("result_table",resultTable);
      //  tenv.toAppendStream(resultTable, Row.class).print("reslut_table===>");//toAppendStream doesn't support consuming update and delete changes
        tenv.toRetractStream(resultTable, Row.class).print("result_table=====>");
        //创建upsert-kafka 列表
        tenv.executeSql("create table dwd_order_pre(\n" +
                "    id String,\n" +
                "    order_id String,\n" +
                "    sku_id String,\n" +
                "    sku_name String,\n" +
                "    order_price String,\n" +
                "    sku_num String,\n" +
                "    create_time String,\n" +
                "    source_type_id String,\n" +
                "    source_type_name String,\n" +
                "    `source_id` string,\n " +
                "    split_total_amount String,\n" +
                "    split_activity_amount String,\n" +
                "    split_coupon_amount String,\n" +
                "    consignee String,\n" +
                "    consignee_tel String,\n" +
                "    total_amount String,\n" +
                "    order_status String,\n" +
                "    user_id String,\n" +
                "    payment_way String,\n" +
                "    delivery_address String,\n" +
                "    order_comment String,\n" +
                "    out_trade_no String,\n" +
                "    trade_body String,\n" +
                "    oprate_time String,\n" +
                "    expire_time String,\n" +
                "    process_status String,\n" +
                "    tracking_no String,\n" +
                "    parent_order_id String,\n" +
                "    province_id String,\n" +
                "    activity_reduce_amount String,\n" +
                "    coupon_reduce_amount String,\n" +
                "    original_total_amount String,\n" +
                "    feight_fee String,\n" +
                "    feight_fee_reduce String,\n" +
                "    refundable_time String,\n" +
                "    order_detail_activity_id String,\n" +
                "    activity_id String,\n" +
                "    activity_rule_id String,\n" +
                "    order_detail_coupon_id String,\n" +
                "    coupon_id String,\n" +
                "    coupon_use_id String,\n" +
                "    `type` String,\n" +
                "    `old` Map<String,String>,\n" +
                "    row_op_ts  TIMESTAMP_LTZ(3),\n" +
                "    primary key(id) not enforced\n" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));
        tenv.executeSql("insert into dwd_order_pre select * from result_table");
        env.execute("DwdTradeOrderPreProcess");

    }
}
